package net.bwie.realtime.jtp.dwd.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.io.IOException;
import java.io.Serializable;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.Period;
import java.util.*;

/**
 * 达摩盘用户画像标签处理 - 增强版（包含年龄和体重标签）
 */
public class UserProfileTagsAgeWeightJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 从Kafka消费用户画像原始数据
        DataStream<String> userProfileDataStream = KafkaUtil.consumerKafka(env, "user_profile_tags");

        // 清理日志并处理
        DataStream<JSONObject> processedStream = processUserProfiles(userProfileDataStream);

        // 将处理结果发送到输出主题
        SingleOutputStreamOperator<String> outputStream = processedStream.map(JSON::toString);

//        outputStream.print("年龄输出");
//        outputStream.print("体重输出");
//        outputStream.print("身高输出");
//        outputStream.print("性别输出");

//        KafkaUtil.producerKafka(outputStream, "user_profile_Age_tags");
//        KafkaUtil.producerKafka(outputStream, "user_profile_Weight_tags");
//        KafkaUtil.producerKafka(outputStream, "user_profile_Height_tags");
//        KafkaUtil.producerKafka(outputStream, "user_profile_Gender_tags");

        env.execute("EnhancedUserProfileTagsJob-" + System.currentTimeMillis());
    }

    /**
     * 用户画像处理流程
     */
    private static DataStream<JSONObject> processUserProfiles(DataStream<String> userProfileDataStream) {
        // 1. 日志清洗
        DataStream<String> cleanedStream = logCleaned(userProfileDataStream);

        // 2. 转换JSON字符串为JSONObject对象（保留所有原始字段）
        DataStream<JSONObject> parsedStream = cleanedStream.map(
                        (MapFunction<String, JSONObject>) JSON::parseObject)
                .name("parse-user-profile");

        // 3. 应用年龄和体重标签处理（带状态）
        DataStream<JSONObject> processedStream = parsedStream
                .keyBy(json -> json.getInteger("user_id"))
                .process(new AgeWeightLabelProcessorWithState())
                .name("age-weight-label-processor");

//        processedStream.print();

        return processedStream;
    }

    /**
     * 日志清洗：过滤无效JSON数据
     */
    private static DataStream<String> logCleaned(DataStream<String> userProfileDataStream) {
        OutputTag<String> dirtyTag = new OutputTag<String>("dirty_log"){};

        SingleOutputStreamOperator<String> cleanedStream = userProfileDataStream.process(
                new ProcessFunction<String, String>() {
                    @Override
                    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                        try {
                            JSON.parseObject(value);
                            out.collect(value);
                        } catch (Exception e) {
                            ctx.output(dirtyTag, value);
                        }
                    }
                }
        );

        // 处理脏数据
        DataStream<String> dirtyStream = cleanedStream.getSideOutput(dirtyTag);
        KafkaUtil.producerKafka(dirtyStream, "user_profile_dirty_tags");

        return cleanedStream;
    }

    /**
     * 带状态的年龄和体重标签处理器
     */
    public static class AgeWeightLabelProcessorWithState extends KeyedProcessFunction<Integer, JSONObject, JSONObject> {
        // 年龄分段定义
        private static final String AGE_18_24 = "18-24岁";
        private static final String AGE_25_29 = "25-29岁";
        private static final String AGE_30_34 = "30-34岁";
        private static final String AGE_35_39 = "35-39岁";
        private static final String AGE_40_49 = "40-49岁";
        private static final String AGE_50_PLUS = "50岁以上";

        // 维度权重
        private static final double CATEGORY_WEIGHT = 0.30;
        private static final double BRAND_WEIGHT = 0.20;
        private static final double PRICE_SENSITIVITY_WEIGHT = 0.15;
        private static final double TIME_BEHAVIOR_WEIGHT = 0.10;
        private static final double SEARCH_TERMS_WEIGHT = 0.10;
        private static final double SOCIAL_INTERACTION_WEIGHT = 0.10;
        private static final double DEVICE_WEIGHT = 0.05;

        // 体重数据来源权重
        private static final double HEALTH_DEVICE_WEIGHT = 1.0;
        private static final double ORDER_INFO_WEIGHT = 0.8;
        private static final double MEMBER_PROFILE_WEIGHT = 0.6;
        private static final double ACTIVITY_FORM_WEIGHT = 0.4;
        //身高 范围
        private static final double MIN_HEIGHT = 100.0;
        private static final double MAX_HEIGHT = 250.0;

        // 状态描述符
        private transient ValueState<UserScoringRules> rulesState;
        private transient ValueState<Double> cumulativeScoreState;
        private transient ValueState<Double> weightedWeightState;

        // 初始规则集
        private final Map<String, Integer> initialCategoryScores = new HashMap<>();
        private final Map<String, Integer> initialBrandScores = new HashMap<>();
        private final Map<String, Integer> initialSearchTermScores = new HashMap<>();
        private final Map<String, Integer> initialDeviceScores = new HashMap<>();

        public AgeWeightLabelProcessorWithState() {
            // 初始化默认规则集
            initDefaultScoringRules();
        }

        private void initDefaultScoringRules() {
            // 类目偏好评分规则
            initialCategoryScores.put("数码电子", 85);
            initialCategoryScores.put("美妆护肤", 80);
            initialCategoryScores.put("运动户外", 75);
            initialCategoryScores.put("母婴用品", 45);
            initialCategoryScores.put("保健品", 40);
            initialCategoryScores.put("金融理财", 35);
            initialCategoryScores.put("退休规划", 20);
            initialCategoryScores.put("汽车用品", 40);
            initialCategoryScores.put("食品生鲜", 50);
            initialCategoryScores.put("服装鞋包", 65);

            // 品牌偏好评分规则
            initialBrandScores.put("设计师品牌", 90);
            initialBrandScores.put("潮牌", 85);
            initialBrandScores.put("奢侈品牌", 70);
            initialBrandScores.put("大众品牌", 50);
            initialBrandScores.put("经典品牌", 40);
            initialBrandScores.put("国货品牌", 60);
            initialBrandScores.put("国际大牌", 70);
            initialBrandScores.put("轻奢品牌", 75);

            // 搜索词评分规则
            initialSearchTermScores.put("电竞", 90);
            initialSearchTermScores.put("追星", 85);
            initialSearchTermScores.put("抗初老", 80);
            initialSearchTermScores.put("轻奢", 75);
            initialSearchTermScores.put("育儿经验", 45);
            initialSearchTermScores.put("退休规划", 30);
            initialSearchTermScores.put("考研资料", 60);
            initialSearchTermScores.put("健身器材", 65);
            initialSearchTermScores.put("抗衰老", 40);
            initialSearchTermScores.put("零食", 55);

            // 设备评分规则
            initialDeviceScores.put("iphone", 85);
            initialDeviceScores.put("pro", 80);
            initialDeviceScores.put("安卓旗舰", 75);
            initialDeviceScores.put("小米", 65);
            initialDeviceScores.put("华为", 70);
            initialDeviceScores.put("mac", 75);
            initialDeviceScores.put("三星", 70);
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);

            ValueStateDescriptor<UserScoringRules> rulesDescriptor =
                    new ValueStateDescriptor<>("scoringRules", UserScoringRules.class);
            rulesState = getRuntimeContext().getState(rulesDescriptor);

            ValueStateDescriptor<Double> cumulativeScoreDescriptor =
                    new ValueStateDescriptor<>("cumulativeScore", Double.class);
            cumulativeScoreState = getRuntimeContext().getState(cumulativeScoreDescriptor);

            ValueStateDescriptor<Double> weightedWeightDescriptor =
                    new ValueStateDescriptor<>("weightedWeight", Double.class);
            weightedWeightState = getRuntimeContext().getState(weightedWeightDescriptor);
        }


        @Override
        public void processElement(JSONObject profile, Context ctx, Collector<JSONObject> out) throws Exception {
            if (rulesState.value() == null) {
                UserScoringRules defaultRules = new UserScoringRules(
                        new HashMap<>(initialCategoryScores),
                        new HashMap<>(initialBrandScores),
                        new HashMap<>(initialSearchTermScores),
                        new HashMap<>(initialDeviceScores)
                );
                rulesState.update(defaultRules);
            }

            // 更新时间戳
            profile.put("process_time", LocalDateTime.now().toString());

            // 年龄
            {
                // 处理年龄标签并获取输出对象
//                JSONObject outputObject = processAgeLabel(profile);
//
//                // 如果有输出对象，则发送
//                if (outputObject != null && !outputObject.isEmpty()) {
//                    out.collect(outputObject);
//                } else {
//                    // 如果没有输出对象，至少发送更新后的profile
//                    out.collect(profile);
//                }
            }

//          体重
            {
//            {
//                // 处理体重标签并获取输出对象
//                JSONObject outputObject = processWeightLabel(profile);
//
//                // 输出结果
//                if (outputObject != null && !outputObject.isEmpty()) {
//                    out.collect(outputObject);
//                }
//            }
            }
////         身高
            {
//            {
//            // 在处理流程中调用
//            JSONObject outputObject = processHeightLabel(profile);
//            if (outputObject != null && !outputObject.isEmpty()) {
//                out.collect(outputObject);
////                }
//            }
            }
//          性别
            {
//            JSONObject genderOutput = processGenderLabel(profile);
//            if (genderOutput != null && !genderOutput.isEmpty()) {
//                out.collect(genderOutput);
//            }
            }
//        }
            // 更新规则集
            updateScoringRules(profile);
        }

        private JSONObject processGenderLabel(JSONObject profile) throws Exception {
            // 创建输出对象
            JSONObject output = new JSONObject();
            output.put("user_id", profile.getInteger("user_id"));
            output.put("process_time", LocalDateTime.now().toString());

            // 1. 检查顶层gender是否有效
            String topLevelGender = profile.getString("gender");
            if (isGenderValid(topLevelGender)) {
                output.put("source", "top_level");
                output.put("gender", topLevelGender);
                output.put("source_weight", 1.0);
                return output;
            }

            // 2. 基于用户行为计算性别
            GenderResult calculatedGender = calculateGenderFromBehavior(profile);
            if (calculatedGender != null && isGenderValid(calculatedGender.gender)) {
                output.put("source", "behavior_analysis");
                output.put("gender", calculatedGender.gender);
                output.put("source_weight", calculatedGender.confidence);
                return output;
            }

            // 3. 检查状态中的有效值（如果需要可以添加状态管理）

            // 4. 无有效值时的处理
            output.put("source", "none");
            output.put("gender", null);
            output.put("source_weight", 0.0);

            return output;
        }

        // 辅助方法：检查性别是否有效
        private boolean isGenderValid(String gender) {
            return gender != null && (gender.equals("男性用户") || gender.equals("女性用户") || gender.equals("家庭用户"));
        }

        // 基于用户行为计算性别
        private GenderResult calculateGenderFromBehavior(JSONObject profile) {
            // 获取用户行为数据
            int purchaseCount = profile.getInteger("purchase_count") != null ? profile.getInteger("purchase_count") : 0;
            int cartCount = profile.getInteger("cart_count") != null ? profile.getInteger("cart_count") : 0;
            int browseCount = profile.getInteger("browse_count") != null ? profile.getInteger("browse_count") : 0;

            // 获取类目偏好
            String categoryPref = profile.getString("category_pref");

            // 计算行为总分（购买50%，加购30%，浏览20%）
            double totalScore = purchaseCount * 0.5 + cartCount * 0.3 + browseCount * 0.2;
            if (totalScore == 0) {
                return null;
            }

            // 根据类目偏好确定性别倾向
            double maleScore = 0;
            double femaleScore = 0;
            double familyScore = 0;

            // 类目性别倾向权重（实际应用中可以根据业务需求调整）
            switch(categoryPref) {
                case "美妆护肤":
                case "母婴用品":
                    femaleScore = totalScore * 0.9;
                    familyScore = totalScore * 0.1;
                    break;
                case "数码电子":
                case "汽车用品":
                    maleScore = totalScore * 0.9;
                    break;
                case "运动户外":
                    maleScore = totalScore * 0.6;
                    femaleScore = totalScore * 0.4;
                    break;
                case "服装鞋包":
                    femaleScore = totalScore * 0.7;
                    maleScore = totalScore * 0.3;
                    break;
                case "食品生鲜":
                case "家居用品":
                    familyScore = totalScore * 0.8;
                    femaleScore = totalScore * 0.2;
                    break;
                default:
                    maleScore = totalScore * 0.5;
                    femaleScore = totalScore * 0.5;
            }

            // 确定最终性别标签
            String gender;
            double confidence;

            if (familyScore > maleScore && familyScore > femaleScore) {
                gender = "家庭用户";
                confidence = familyScore / totalScore;
            } else if (femaleScore > maleScore) {
                gender = "女性用户";
                confidence = femaleScore / totalScore;
            } else {
                gender = "男性用户";
                confidence = maleScore / totalScore;
            }

            return new GenderResult(gender, confidence);
        }

        // 性别计算结果内部类
        private static class GenderResult {
            String gender;
            double confidence;

            GenderResult(String gender, double confidence) {
                this.gender = gender;
                this.confidence = confidence;
            }
        }

        private GenderScores calculateGenderScores(int purchaseCount, int cartCount, int browseCount, String categoryPref) {
            GenderScores scores = new GenderScores();

            // Behavior weights: purchase (50%), cart/favorite (30%), browse (20%)
            double totalBehaviors = purchaseCount + cartCount + browseCount;
            if (totalBehaviors > 0) {
                // Calculate category weights (simplified - in reality would use more detailed category mapping)
                double femaleWeight = 0;
                double maleWeight = 0;
                double familyWeight = 0;

                // Simplified category mapping (in production would use a more comprehensive mapping)
                switch (categoryPref) {
                    case "美妆护肤":
                    case "母婴用品":
                        femaleWeight = 1.0;
                        break;
                    case "数码电子":
                    case "汽车用品":
                        maleWeight = 1.0;
                        break;
                    case "运动户外":
                    case "食品生鲜":
                        // Neutral/slightly family oriented
                        familyWeight = 0.7;
                        femaleWeight = 0.15;
                        maleWeight = 0.15;
                        break;
                    case "服装鞋包":
                        // Could be either
                        femaleWeight = 0.6;
                        maleWeight = 0.4;
                        break;
                    default:
                        familyWeight = 0.5;
                        femaleWeight = 0.25;
                        maleWeight = 0.25;
                }

                // Apply behavior weights to scores
                scores.femaleScore = (purchaseCount * 0.5 + cartCount * 0.3 + browseCount * 0.2) * femaleWeight;
                scores.maleScore = (purchaseCount * 0.5 + cartCount * 0.3 + browseCount * 0.2) * maleWeight;
                scores.familyScore = (purchaseCount * 0.5 + cartCount * 0.3 + browseCount * 0.2) * familyWeight;
            }

            return scores;
        }

        private String determineGenderLabel(GenderScores scores) {
            double total = scores.femaleScore + scores.maleScore + scores.familyScore;
            if (total == 0) {
                return "未知"; // Default if no data
            }

            // Check for family user condition first
            double familyRatio = scores.familyScore / total;
            double genderDiff = Math.abs(scores.femaleScore - scores.maleScore) / total;

            if (familyRatio >= 0.3 && genderDiff <= 0.1) {
                return "家庭用户";
            }

            // Otherwise return the dominant gender
            if (scores.femaleScore > scores.maleScore) {
                return "女性用户";
            } else {
                return "男性用户";
            }
        }

        // Helper class to store gender scores
        private static class GenderScores {
            double femaleScore;
            double maleScore;
            double familyScore;
        }

        private static final String[] ZODIAC_SIGNS = {
                "摩羯座", "水瓶座", "双鱼座", "白羊座", "金牛座", "双子座",
                "巨蟹座", "狮子座", "处女座", "天秤座", "天蝎座", "射手座", "摩羯座"
        };

        // 星座分界日
        private static final int[] ZODIAC_BOUNDARIES = {
                20, 19, 21, 20, 21, 22, 23, 23, 23, 24, 23, 22, 20
        };

        // 计算星座方法
        private String calculateZodiacSign(int month, int day) {
            if (month < 1 || month > 12 || day < 1 || day > 31) {
                return "未知";
            }
            return day < ZODIAC_BOUNDARIES[month-1] ? ZODIAC_SIGNS[month-1] : ZODIAC_SIGNS[month];
        }


        private JSONObject processAgeLabel(JSONObject profile) throws Exception {
            // 创建只包含所需字段的输出对象
            JSONObject output = new JSONObject();
            output.put("user_id", profile.getInteger("user_id"));
            output.put("category_pref", profile.getString("category_pref"));
            output.put("brand_pref", profile.getString("brand_pref"));
            output.put("price_sensitivity", profile.getString("price_sensitivity"));
            output.put("time_behavior", profile.getString("time_behavior"));
            output.put("search_terms", profile.getString("search_terms"));
            output.put("social_interaction", profile.getString("social_interaction"));
            output.put("device_info", profile.getString("device_info"));
            output.put("constellation", profile.getString("constellation"));
            output.put("process_time", LocalDateTime.now().toString());

            // 优先使用完整生日计算
            Integer birthYear = parseBirthField(profile.getString("birth_year"));
            Integer birthMonth = parseBirthField(profile.getString("birth_month"));
            Integer birthDay = parseBirthField(profile.getString("birth_day"));

            boolean birthDateChanged = false;

            // 检查生日是否被修改
            if (isValidBirthDate(birthYear, birthMonth, birthDay)) {
                String currentBirthKey = birthYear + "-" + birthMonth + "-" + birthDay;
                String lastBirthKey = profile.getString("last_birth_key");

                if (!currentBirthKey.equals(lastBirthKey)) {
                    birthDateChanged = true;
                    profile.put("last_birth_key", currentBirthKey);
                }

                int age = calculateAge(birthYear, birthMonth, birthDay);
                output.put("age_label", getAgeBucketByAge(age));

                // 如果有生日变更，重置累积分数
                if (birthDateChanged) {
                    cumulativeScoreState.clear();
                }
                return output;
            }

            // 累积得分计算
            double currentScore = calculateWeightedAgeScore(profile);

            // 获取状态中的累积分数
            Double cumulativeScore = cumulativeScoreState.value();
            if (cumulativeScore == null) {
                cumulativeScore = 0.0;
            }

            // 基于最新数据更新累积分数（指数平滑）
            cumulativeScore = 0.3 * currentScore + 0.7 * cumulativeScore;
            cumulativeScoreState.update(cumulativeScore);

            // 设置年龄标签
            output.put("age_label", getAgeBucketByScore(cumulativeScore));

            return output;
        }

        private JSONObject processWeightLabel(JSONObject profile) throws Exception {
            // 创建输出对象
            JSONObject output = new JSONObject();
            output.put("user_id", profile.getInteger("user_id"));
            output.put("process_time", LocalDateTime.now().toString());

            // 1. 检查顶层weight是否有效
            Double topLevelWeight = parseWeight(profile.get("weight"));
            if (isWeightValid(topLevelWeight)) {
                profile.put("weight", topLevelWeight);
                weightedWeightState.update(topLevelWeight);

                output.put("source", "top_level");
//                output.put("source_label", "顶层weight字段");
                output.put("weight", topLevelWeight);
                output.put("source_weight", 1.0);
                return output;
            }

            // 2. 检查weight_records
            JSONArray weightRecords = profile.getJSONArray("weight_records");
            JSONObject orderInfoRecord = null;

            if (weightRecords != null && !weightRecords.isEmpty()) {
                // 按优先级排序
                List<JSONObject> sortedRecords = new ArrayList<>();
                for (int i = 0; i < weightRecords.size(); i++) {
                    JSONObject record = weightRecords.getJSONObject(i);
                    sortedRecords.add(record);

                    // 记录order_info信息
                    if ("order_info".equals(record.getString("source"))) {
                        orderInfoRecord = record;
                    }
                }

                sortedRecords.sort((a, b) -> Double.compare(
                        b.getDoubleValue("source_weight"),
                        a.getDoubleValue("source_weight")));

                // 寻找第一个有效记录
                for (JSONObject record : sortedRecords) {
                    Double weight = parseWeight(record.get("weight"));
                    if (isWeightValid(weight)) {
                        profile.put("weight", weight);
                        weightedWeightState.update(weight);

                        output.put("source", record.getString("source"));
                        output.put("source_label", record.getString("source_label"));
                        output.put("weight", weight);
                        output.put("source_weight", record.getDouble("source_weight"));
                        return output;
                    }
                }
            }

            // 3. 检查状态中的有效值
            Double lastValidWeight = weightedWeightState.value();
            if (isWeightValid(lastValidWeight)) {
                profile.put("weight", lastValidWeight);

                output.put("source", "state");
                output.put("source_label", "状态存储的上次有效值");
                output.put("weight", lastValidWeight);
                output.put("source_weight", 1.0);
                return output;
            }

            // 4. 无有效值时的处理
            profile.put("weight", null);

            // 如果有order_info记录，即使无效也输出
            if (orderInfoRecord != null) {
                Double weight = parseWeight(orderInfoRecord.get("weight"));
                output.put("source", orderInfoRecord.getString("source"));
                output.put("source_label", orderInfoRecord.getString("source_label"));
                output.put("weight", weight);
                output.put("source_weight", orderInfoRecord.getDouble("source_weight"));
            } else {
                output.put("source", "none");
                output.put("source_label", "无有效数据");
                output.put("weight", null);
                output.put("source_weight", 0.0);
            }

            return output;
        }

        // 辅助方法：检查体重是否有效
        private boolean isWeightValid(Double weight) {
            return weight != null && weight >= 30 && weight <= 200;
        }

        private Double parseWeight(Object weightObj) {
            if (weightObj == null) return null;

            try {
                if (weightObj instanceof Number) {
                    return ((Number) weightObj).doubleValue();
                } else if (weightObj instanceof String) {
                    String weightStr = ((String) weightObj).replaceAll("[^0-9.]", "");
                    if (weightStr.isEmpty()) return null;
                    return Double.parseDouble(weightStr);
                }
                return null;
            } catch (NumberFormatException e) {
                return null;
            }
        }

        private JSONObject processHeightLabel(JSONObject profile) throws Exception {
            // 创建输出对象
            JSONObject output = new JSONObject();
            output.put("user_id", profile.getInteger("user_id"));
            output.put("process_time", LocalDateTime.now().toString());

            // 1. 检查顶层height是否有效
            Double topLevelHeight = parseHeight(profile.get("height"));
            if (isHeightValid(topLevelHeight)) {
                profile.put("height", topLevelHeight);
                weightedWeightState.update(topLevelHeight);

                output.put("source", "top_level");
                output.put("source_label", "顶层height字段");
                output.put("height", topLevelHeight);
                output.put("source_weight", 1.0);
                return output;
            }

            // 2. 检查height_records
            JSONArray heightRecords = profile.getJSONArray("height_records");
            JSONObject orderInfoRecord = null;

            if (heightRecords != null && !heightRecords.isEmpty()) {
                // 按优先级排序
                List<JSONObject> sortedRecords = new ArrayList<>();
                for (int i = 0; i < heightRecords.size(); i++) {
                    JSONObject record = heightRecords.getJSONObject(i);
                    sortedRecords.add(record);

                    // 记录order_info信息
                    if ("order_info".equals(record.getString("source"))) {
                        orderInfoRecord = record;
                    }
                }

                sortedRecords.sort((a, b) -> Double.compare(
                        b.getDoubleValue("source_weight"),
                        a.getDoubleValue("source_weight")));

                // 寻找第一个有效记录
                for (JSONObject record : sortedRecords) {
                    Double height = parseHeight(record.get("height"));
                    if (isHeightValid(height)) {
                        profile.put("height", height);
                        weightedWeightState.update(height);

                        output.put("source", record.getString("source"));
                        output.put("source_label", record.getString("source_label"));
                        output.put("height", height);
                        output.put("source_weight", record.getDouble("source_weight"));
                        return output;
                    }
                }
            }

            // 3. 检查状态中的有效值
            Double lastValidHeight = weightedWeightState.value();
            if (isHeightValid(lastValidHeight)) {
                profile.put("height", lastValidHeight);

                output.put("source", "state");
                output.put("source_label", "状态存储的上次有效值");
                output.put("height", lastValidHeight);
                output.put("source_weight", 1.0);
                return output;
            }

            // 4. 无有效值时的处理
            profile.put("height", null);

            // 如果有order_info记录，即使无效也输出
            if (orderInfoRecord != null) {
                Double height = parseHeight(orderInfoRecord.get("height"));
                output.put("source", orderInfoRecord.getString("source"));
                output.put("source_label", orderInfoRecord.getString("source_label"));
                output.put("height", height);
                output.put("source_weight", orderInfoRecord.getDouble("source_weight"));
            } else {
                output.put("source", "none");
                output.put("source_label", "无有效数据");
                output.put("height", null);
                output.put("source_weight", 0.0);
            }

            return output;
        }

        // 辅助方法：检查身高是否有效
        private boolean isHeightValid(Double height) {
            return height != null && height >= MIN_HEIGHT && height <= MAX_HEIGHT;
        }

        // 添加身高解析方法
        private Double parseHeight(Object heightObj) {
            if (heightObj == null) return null;

            try {
                if (heightObj instanceof Number) {
                    return ((Number) heightObj).doubleValue();
                } else if (heightObj instanceof String) {
                    String heightStr = ((String) heightObj).replaceAll("[^0-9.]", "");
                    if (heightStr.isEmpty()) return null;
                    return Double.parseDouble(heightStr);
                }
                return null;
            } catch (NumberFormatException e) {
                return null;
            }
        }

        private boolean isWeightInNormalRange(Double weight) {
            return weight != null && weight >= 30 && weight <= 200;
        }

        private Integer parseBirthField(String value) {
            if (StringUtils.isBlank(value) || "".equals(value.trim())) {
                return null;
            }
            try {
                return Integer.parseInt(value);
            } catch (NumberFormatException e) {
                return null;
            }
        }

        private boolean isValidBirthDate(Integer birthYear, Integer birthMonth, Integer birthDay) {
            if (birthYear == null || birthMonth == null || birthDay == null)
                return false;

            int currentYear = LocalDate.now().getYear();
            if (birthYear < 1900 || birthYear > currentYear)
                return false;
            if (birthMonth < 1 || birthMonth > 12)
                return false;
            if (birthDay < 1 || birthDay > 31)
                return false;

            // 闰年检查
            if (birthMonth == 2) {
                boolean isLeap = (birthYear % 4 == 0 && (birthYear % 100 != 0 || birthYear % 400 == 0));
                int maxDay = isLeap ? 29 : 28;
                if (birthDay > maxDay) return false;
            }
            // 小月检查
            else if (Arrays.asList(4, 6, 9, 11).contains(birthMonth)) {
                if (birthDay > 30) return false;
            }

            return true;
        }

        private int calculateAge(int birthYear, int birthMonth, int birthDay) {
            LocalDate birthDate = LocalDate.of(birthYear, birthMonth, birthDay);
            LocalDate currentDate = LocalDate.now();
            return Period.between(birthDate, currentDate).getYears();
        }

        private String getAgeBucketByAge(int age) {
            if (age < 18) return AGE_18_24; // 特殊处理未成年用户
            if (age <= 24) return AGE_18_24;
            if (age <= 29) return AGE_25_29;
            if (age <= 34) return AGE_30_34;
            if (age <= 39) return AGE_35_39;
            if (age <= 49) return AGE_40_49;
            return AGE_50_PLUS;
        }

        private double calculateWeightedAgeScore(JSONObject profile) throws IOException {

            double totalScore = 0;

            // 获取状态中的评分规则
            UserScoringRules rules = rulesState.value();
            if (rules == null) {
                rules = new UserScoringRules(
                        new HashMap<>(initialCategoryScores),
                        new HashMap<>(initialBrandScores),
                        new HashMap<>(initialSearchTermScores),
                        new HashMap<>(initialDeviceScores)
                );
            }

            // 1. 类目偏好维度 (30%)
            String category = profile.getString("category_pref");
            int categoryScore = rules.categoryScores.getOrDefault(category, 50);
            totalScore += categoryScore * CATEGORY_WEIGHT;

            // 2. 品牌偏好维度 (20%)
            String brand = profile.getString("brand_pref");
            int brandScore = rules.brandScores.getOrDefault(brand, 50);
            totalScore += brandScore * BRAND_WEIGHT;

            // 3. 价格敏感度维度 (15%)
            Double priceSensitivity = profile.getDouble("price_sensitivity");
            if (priceSensitivity == null) priceSensitivity = 0.5;
            totalScore += (priceSensitivity * 100) * PRICE_SENSITIVITY_WEIGHT;

            // 4. 时间行为维度 (10%)
            JSONObject timeBehavior = profile.getJSONObject("time_behavior");
            if (timeBehavior != null) {
                Integer nightCount = timeBehavior.getInteger("night");
                String activityLevel = timeBehavior.getString("recent_activity_level");

                double timeScore = 0;
                if (nightCount != null) {
                    timeScore += Math.min(nightCount * 5, 70);
                }

                // 根据活动水平增加分数
                if ("high".equalsIgnoreCase(activityLevel)) {
                    timeScore += 20;
                } else if ("medium".equalsIgnoreCase(activityLevel)) {
                    timeScore += 10;
                }

                totalScore += Math.min(timeScore, 100) * TIME_BEHAVIOR_WEIGHT;
            } else {
                totalScore += 50 * TIME_BEHAVIOR_WEIGHT;
            }

            // 5. 搜索词分析维度 (10%)
            int searchScore = calculateSearchTermScore(profile.getString("search_terms"), rules.searchTermScores);
            totalScore += searchScore * SEARCH_TERMS_WEIGHT;

            // 6. 社交互动行为 (10%)
            Integer social = profile.getInteger("social_interaction");
            if (social != null) {
                // 非线性转换
                double socialScore;
                if (social > 80) socialScore = 90;
                else if (social > 60) socialScore = 75;
                else if (social > 40) socialScore = 55;
                else if (social > 20) socialScore = 40;
                else socialScore = 25;

                totalScore += socialScore * SOCIAL_INTERACTION_WEIGHT;
            } else {
                totalScore += 50 * SOCIAL_INTERACTION_WEIGHT;
            }

            // 7. 设备信息维度 (5%)
            int deviceScore = getDeviceScore(profile.getString("device_info"), rules.deviceScores);
            totalScore += deviceScore * DEVICE_WEIGHT;

            return totalScore;
        }

        private int calculateSearchTermScore(String searchTerms, Map<String, Integer> searchTermScores) {
            if (StringUtils.isBlank(searchTerms)) return 50;

            int total = 0;
            int count = 0;
            String[] terms = searchTerms.split(" ");

            for (String term : terms) {
                term = term.trim();
                if (term.isEmpty()) continue;

                Integer score = searchTermScores.get(term);
                if (score != null) {
                    total += score;
                    count++;
                } else {
                    // 部分匹配
                    for (Map.Entry<String, Integer> entry : searchTermScores.entrySet()) {
                        if (term.contains(entry.getKey())) {
                            total += entry.getValue();
                            count++;
                            break;
                        }
                    }
                }
            }

            return count > 0 ? total / count : 50;
        }

        private int getDeviceScore(String deviceInfo, Map<String, Integer> deviceScores) {
            if (StringUtils.isBlank(deviceInfo)) return 55;

            String deviceLower = deviceInfo.toLowerCase();

            // 首先尝试全匹配
            Integer directScore = deviceScores.get(deviceLower);
            if (directScore != null) return directScore;

            // 否则尝试关键字匹配
            for (Map.Entry<String, Integer> entry : deviceScores.entrySet()) {
                if (deviceLower.contains(entry.getKey())) {
                    return entry.getValue();
                }
            }

            return 55;
        }

        private String getAgeBucketByScore(double score) {
            if (score >= 85) return AGE_18_24;
            if (score >= 70) return AGE_25_29;
            if (score >= 55) return AGE_30_34;
            if (score >= 45) return AGE_35_39;
            if (score >= 30) return AGE_40_49;
            return AGE_50_PLUS;
        }

        private void updateScoringRules(JSONObject profile) throws Exception {
            UserScoringRules rules = rulesState.value();

            // 更新类目评分规则
            String category = profile.getString("category_pref");
            if (category != null && !rules.categoryScores.containsKey(category)) {
                // 动态添加新类目的基准分数
                Integer social = profile.getInteger("social_interaction");
                int defaultScore = (social != null && social > 50) ? 65 : 45;
                rules.categoryScores.put(category, defaultScore);
            }

            // 更新品牌评分规则
            String brand = profile.getString("brand_pref");
            if (brand != null && !rules.brandScores.containsKey(brand)) {
                // 动态添加新品牌基准分数
                Double priceSensitivity = profile.getDouble("price_sensitivity");
                int defaultScore = (priceSensitivity != null && priceSensitivity < 0.3) ? 75 : 55;
                rules.brandScores.put(brand, defaultScore);
            }

            // 更新搜索词规则
            String searchTerms = profile.getString("search_terms");
            if (StringUtils.isNotBlank(searchTerms)) {
                for (String term : searchTerms.split(" ")) {
                    if (!rules.searchTermScores.containsKey(term)) {
                        rules.searchTermScores.put(term, 50); // 中性分数
                    }
                }
            }

            rulesState.update(rules);
        }
    }

    /**
     * 用户评分规则状态对象
     */
    public static class UserScoringRules implements Serializable {
        Map<String, Integer> categoryScores;
        Map<String, Integer> brandScores;
        Map<String, Integer> searchTermScores;
        Map<String, Integer> deviceScores;

        public UserScoringRules() {
            // 默认构造函数（用于序列化）
        }

        public UserScoringRules(
                Map<String, Integer> categoryScores,
                Map<String, Integer> brandScores,
                Map<String, Integer> searchTermScores,
                Map<String, Integer> deviceScores) {

            this.categoryScores = categoryScores;
            this.brandScores = brandScores;
            this.searchTermScores = searchTermScores;
            this.deviceScores = deviceScores;
        }
    }
}